/*
 * Copyright 2009 Armando Blancas
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package taskgraph.tasks.io;

import java.beans.IntrospectionException;
import java.io.IOException;
import java.io.InterruptedIOException;
import java.lang.reflect.InvocationTargetException;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;

import taskgraph.ports.OutputPort;
import taskgraph.reflect.Accessor;
import taskgraph.reflect.BeanMetadata;
import taskgraph.tasks.PrimitiveTask;


/**
 * Source task for reading from a JDBC Result Set.
 *  
 * @author Armando Blancas
 * @see taskgraph.ports.OutputPort
 * @see java.lang.Class
 * @param <E> The type of elements read by this task.
 */
public class JDBCResultSet<E> extends PrimitiveTask {

	private OutputPort<E> output;
	private Class<E> clazz;
	private String sqlURL;
	private String sqlUser;
	private String sqlPassword;
	private String sqlStatement;

	/**
	 * This constructor allows creating instances as beans.
	 */
    public JDBCResultSet() {
    	// do nothing
    }

    /**
     * JDBCResultSet constructor.
     * 
     * @param output       The element output port.
     * @param clazz        The class literal of type parameter E.
     * @param sqlURL       The JDBC URL that identifies the database.
     * @param sqlStatement The SQL statement to run to get a result set.
     */
    public JDBCResultSet(final OutputPort<E> output,
    	                 final Class<E> clazz,
    	                 final String sqlURL,
    	                 final String sqlStatement) {
    	setOutput(output);
    	setClazz(clazz);
    	setSqlURL(sqlURL);
    	setSqlStatement(sqlStatement);
    }
    
    /**
     * JDBCResultSet constructor.
     * 
     * @param output       The element output port.
     * @param clazz        The class literal of type parameter E.
     * @param sqlURL       The JDBC URL that identifies the database.
     * @param sqlUser      The database user account name.
     * @param sqlPassword  The database user account password.
     * @param sqlStatement The SQL statement to run to get a result set.
     */
    public JDBCResultSet(final OutputPort<E> output,
    	                 final Class<E> clazz,
    	                 final String sqlURL,
    	                 final String sqlUser,
    	                 final String sqlPassword,
    	                 final String sqlStatement) {
    	setOutput(output);
    	setClazz(clazz);
    	setSqlURL(sqlURL);
    	setSqlUser(sqlUser);
    	setSqlPassword(sqlPassword);
    	setSqlStatement(sqlStatement);
    }
    
	/**
	 * Gets the class object for the type parameter E.
	 * 
	 * @return The {@code Class<E>} value.
	 */
	public Class<E> getClazz() {
		return clazz;
	}

	/**
     * Gets the output port.
     * 
	 * @return The element output port.
	 */
	public OutputPort<E> getOutput() {
		return output;
	}

	/**
	 * Gets the database account password.
	 * 
	 * @return The password for the give user.
	 */
	public String getSqlPassword() {
		return sqlPassword;
	}

	/**
     * Gets the SQL statement to execute.
     * 
	 * @return The statement string value.
	 */
	public String getSqlStatement() {
		return sqlStatement;
	}

	/**
	 * Gets the database identification string.
	 * 
	 * @return The database URL.
	 */
	public String getSqlURL() {
		return sqlURL;
	}

	/**
	 * Gets the database user name.
	 * 
	 * @return The user string value.
	 */
	public String getSqlUser() {
		return sqlUser;
	}

	/**
     * Performs the work of this task.
 	 */
    public void run() {
    	try {
    		Connection connection = 
    			DriverManager.getConnection(sqlURL, sqlUser, sqlPassword);
    		Statement statement = connection.createStatement();
    		ResultSet set = statement.executeQuery(sqlStatement);

    		Accessor[] accessors = BeanMetadata.extractSetters(clazz, set);
    		if (accessors.length == 0) {
                log.warn("JDBCResultSet#run(): No setters defined");
                return;
    		}

    		set.beforeFirst();
    		while (!Thread.currentThread().isInterrupted()) {
    			if (!set.next()) {
    				break;
    			}
    			E element = BeanMetadata.newInstance(clazz, accessors, set);
    			output.write(element);
            }
        } catch (InterruptedIOException e) {
        	// interruption request
		} catch (SQLException e) {
            log.error("JDBCResultSet#run()", e);
        } catch (IOException e) {
            log.error("JDBCResultSet#run()", e);
        } catch (InstantiationException e) {
            log.error("JDBCResultSet#run()", e);
		} catch (IllegalAccessException e) {
            log.error("JDBCResultSet#run()", e);
		} catch (IntrospectionException e) {
            log.error("JDBCResultSet#run()", e);
		} catch (IllegalArgumentException e) {
            log.error("JDBCResultSet#run()", e);
		} catch (InvocationTargetException e) {
            log.error("JDBCResultSet#run()", e);
		} finally {
			output.close();
        }
    }

    /**
	 * Sets the class object for the type parameter E.
	 * 
	 * @param clazz The {@code Class<E>} value.
	 */
	public void setClazz(final Class<E> clazz) {
    	if (clazz == null)
    		throw new IllegalArgumentException("clazz == null");
		this.clazz = clazz;
	}

	/**
	 * Sets the output port.
	 * 
	 * @param output The element output port.
	 */
    public void setOutput(final OutputPort<E> output) {
    	if (output == null)
    		throw new IllegalArgumentException("output == null");
 		this.output = output;
	}

    /**
	 * Gets the database account password for the user name.
	 * 
	 * @param sqlPassword The string password to set.
	 */
	public void setSqlPassword(final String sqlPassword) {
    	if (sqlPassword == null)
    		throw new IllegalArgumentException("sqlPassword == null");
		this.sqlPassword = sqlPassword;
	}

	/**
     * Sets the SQL statement to execute.
	 * 
	 * @param sqlStatement The statement string value.
	 */
	public void setSqlStatement(final String sqlStatement) {
    	if (sqlStatement == null)
    		throw new IllegalArgumentException("sqlStatement == null");
		this.sqlStatement = sqlStatement;
	}

	/**
	 * Sets the database identification string.
	 * 
	 * @param sqlURL The URL string to set.
	 */
	public void setSqlURL(String sqlURL) {
    	if (sqlURL == null)
    		throw new IllegalArgumentException("sqlURL == null");
		this.sqlURL = sqlURL;
	}

	/**
	 * Sets the database user name.
	 * 
	 * @param sqlUser The user string value to set.
	 */
	public void setSqlUser(String sqlUser) {
    	if (sqlUser == null)
    		throw new IllegalArgumentException("sqlUser == null");
		this.sqlUser = sqlUser;
	}

}
